package cn.itcast.flink.cdc;

import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import lombok.SneakyThrows;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 *  * 基于DataStream实现Flink CDC案例演示：从MySQL数据库实时获取数据，打印控制台。
 * @author lilulu
 * @date 2023-04-10 20:50
 */
public class FlinkCdcSqlDemo {
    @SneakyThrows
    public static void main(String[] args) {
        //执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(3000);
        env.setParallelism(1);

        StreamTableEnvironment streamEnv = StreamTableEnvironment.create(env, EnvironmentSettings.newInstance().inStreamingMode().useBlinkPlanner().build());

        // 2. 创建输入表，todo：使用CDC-SQL-MySQL获取数据
        streamEnv.executeSql(
                "CREATE TABLE mysql_binlog (\n" +
                        " id INT,\n" +
                        " name STRING,\n" +
                        " age INT,\n" +
                        " gender STRING,\n" +
                        " PRIMARY KEY(id) NOT ENFORCED\n" +
                        ") WITH (\n" +
                        " 'connector' = 'mysql-cdc',\n" +
                        " 'hostname' = 'node1.itcast.cn',\n" +
                        " 'port' = '3306',\n" +
                        " 'username' = 'root',\n" +
                        " 'password' = '123456',\n" +
                        " 'database-name' = 'db_flink',\n" +
                        " 'table-name' = 'tbl_users',\n" +
                        " 'scan.startup.mode' = 'initial'\n" +
                        ")"
        );

        TableResult tableResult = streamEnv.executeSql("select * from mysql_binlog");
        tableResult.print();
    }
}
